package kk.learn.spark.work._2

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

/**
 * <p>
 *
 * </p>
 *
 * @author KK
 * @since 2021-03-05
 */
object LogAnalysis {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("IpArea")
      .master("local[*]")
      .getOrCreate()
    val sc = spark.sparkContext
    sc.setLogLevel("warn")

    // 读取文件为rdd
    val lines: RDD[String] = sc.textFile("code/data/cdn.txt")

    // 这个rdd需要反复用到， 所以进行缓存
    lines.cache()

    // 1 统计独立的IP数
    val ipCount = lines
      .map(_.split("\\s+", 2)(0).trim()) // 切割成两部分， 第一部分就是IP
        .distinct() // 对IP进行去重
        .count() // 计数

    println(s"独立IP的个数： $ipCount")

    // 2 统计每个视频的独立ip数
    lines
      .map(_.split("\\s+")) // 切割每一行
      .map(array => (if(array(6).indexOf("?")>0) array(6).substring(0, array(6).indexOf("?")) else array(6),
        array(0))) // 去除掉url后对请求参数， 不然会影响判断
      .filter(t => t._1.contains(".mp4"))// 筛选出视频记录
      .groupByKey()// 按照URL， 进行group by
      .mapValues(_.toList.distinct.length)// 统计每个视频独立IP数
      .foreach(println(_))// 输出

    // 3 统计一天中每个小时的流量
    val counts = lines
      .map(_.split("\\s+"))
      .map(array => (array(3).substring(array(3).length() - 8, array(3).length() - 6), array(9).toLong))
      .groupByKey()
      .mapValues(_.sum)
      .mapValues(sum => sum / (1024 * 1024 * 1024) + "GB")
      .collect()
    val tuples = counts.sortBy(t => t._1)
    for (tuple <- tuples) {
      println(tuple)
    }


    spark.close()
  }
}
